package com.lagou.niancheng.rpc.client.netty;

import com.lagou.niancheng.rpc.client.Connection;
import com.lagou.niancheng.rpc.client.ConnectionSource;
import com.lagou.niancheng.rpc.client.netty.handler.RPCResponseDataRequestHandler;
import com.lagou.niancheng.rpc.client.netty.handler.RPCServiceMetaDataRequestHandler;
import com.lagou.niancheng.rpc.common.converter.Converter;
import com.lagou.niancheng.rpc.common.entity.RPCResponseData;
import com.lagou.niancheng.rpc.common.entity.RPCServiceMetaData;
import com.lagou.niancheng.rpc.common.exceptions.SimpleRpcException;
import com.lagou.niancheng.rpc.common.handler.RequestHandlerRegistry;
import com.lagou.niancheng.rpc.common.handler.SimpleRpcProtocolDecoder;
import com.lagou.niancheng.rpc.common.handler.SimpleRpcProtocolEncoder;
import com.lagou.niancheng.rpc.common.handler.SimpleRpcTransferDataHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;

/**
 * netty实现的连接源¬
 *
 * @author 大犬
 * @date 2020/5/1
 */
public class NettyConnectionSource implements ConnectionSource {

    private int workerThreadSize;
    private Converter converter;
    private String host;
    private int port;
    private long timeout;
    private RequestResultRegistry requestResultRegistry;
    private Bootstrap bootstrap;

    public NettyConnectionSource(NettyConnectionConfig nettyConnectionConfig) {
        workerThreadSize = nettyConnectionConfig.getWorkerThreadSize();
        converter = nettyConnectionConfig.getConverter();
        host = nettyConnectionConfig.getHost();
        port = nettyConnectionConfig.getPort();
        timeout = nettyConnectionConfig.getTimeout();
        requestResultRegistry = new RequestResultRegistry();
        bootstrap = new Bootstrap();
        init();
    }

    private void init() {
        // 创建工作线程池
        NioEventLoopGroup nioEventLoopGroup = new NioEventLoopGroup(workerThreadSize, (r) -> {
            Thread thread = new Thread(r);
            thread.setUncaughtExceptionHandler((t, e) -> e.printStackTrace());
            return thread;
        });
        // 注册结果处理器
        RequestHandlerRegistry requestHandlerRegistry = new RequestHandlerRegistry();
        requestHandlerRegistry.register(RPCServiceMetaData.class, new RPCServiceMetaDataRequestHandler(requestResultRegistry));
        requestHandlerRegistry.register(RPCResponseData.class, new RPCResponseDataRequestHandler(requestResultRegistry));
        // 初始化客户端bootStrap
        bootstrap.group(nioEventLoopGroup)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.SO_KEEPALIVE, false)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel channel) throws Exception {
                        channel.pipeline()
                                // 协议处理，将请求数据转换为transferData
                                .addLast(new SimpleRpcProtocolDecoder())
                                // response编码
                                .addLast(new SimpleRpcProtocolEncoder())
                                // 处理请求rpc逻辑
                                .addLast(new SimpleRpcTransferDataHandler(converter, requestHandlerRegistry));
                    }
                });

    }


    @Override
    public Connection getConnection() {
        ChannelFuture connect = bootstrap.connect(host, port);
        try {
            connect.get();
        } catch (Exception e) {
            throw new SimpleRpcException("connect failed", e);
        }
        Channel channel = connect.channel();
        return new NettyConnection(channel, converter, requestResultRegistry, timeout);
    }
}
